package com.exm.collectcodenew.kafka.producer.customInterceptor;


import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;
public class LoggingOffsetInterceptor implements ConsumerInterceptor<String,String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        System.out.println("Before consuming records: " + records);
        // 在这里可以进行一些预处理操作，例如修改消息内容等
        // 例如：修改每条消息的内容
        records.forEach(record -> System.out.println("Original message: " + new String(record.value())));
        // 你可以在这里返回一个新的ConsumerRecords对象来改变原始的记录集
        return records;
    }
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        System.out.println("Committing offsets: " + offsets);
        // 在这里可以执行一些在偏移量提交之后的操作，例如记录偏移量信息等
    }
    @Override
    public void close() {
        // 清理资源等操作
    }
    @Override
    public void configure(Map<String, ?> configs) {
    }
}
